package org.niit.service

import org.apache.spark.sql.{DataFrame, Dataset, Row}
import org.niit.bean.AnswerWithRecommendations
import org.niit.util.SparkUtil

/**
 * Date:2025/6/12
 * Author：Ys
 * Description:
 */
class EDUBatchService {

  import org.apache.spark.sql.functions._

  def dataAnalysis(): Unit = {
    val spark = SparkUtil.takeSpark()

    import spark.implicits._
    //1.加载数据
    val data = spark.read.format("jdbc")
      .option("url", "jdbc:mysql://node1:3306/BD2_2025?useUnicode=true&characterEncoding=utf8")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("user","root")
      .option("password","Niit@123")
      .option("dbtable","edu").load().as[AnswerWithRecommendations]
    //需求一
    hotSubjectCountTop50(data)
    //需求二：
    hotSubjectRecommendTop20(data)

  }
  /*
  需求1:各科目热点题分析要求:找到Top50热点题对应的科目，然后统计这些科目中，分别包含这几道热点题的条目数
  举例:
  热点题
  题号 热度 学科
  1 100 数学
  2 99 数学
  3 98 语文
  最终结果:
  学科 热点题数量
  数学 2
  语文 1
   */
  def hotSubjectCountTop50(data:Dataset[AnswerWithRecommendations]): Unit = {
    //1. 统计前50道热点题--->>>在数据库中，即使相同的题目，也是分布在不同行中的
    // 张三  题目1  数学
    //李四   题目1  数学    ==》 题目1   2
    val hotTop50: Dataset[Row] = data.groupBy("question_id")
      .agg(count("*").as("hot"))
      .orderBy(desc("hot"))
      .limit(50)

    /*
     //2. 将hotTop50结果和data进行关联，得到热点题对应科目
      dropDuplicates：将重复的题目删掉 （去重）
     */
    val joinDF: DataFrame = hotTop50.join(data.dropDuplicates("question_id"), "question_id")


    //3. 按学科分组聚合统计各个学科包含的热点题数量
    val res = joinDF.groupBy("subject_id")
      .agg(count("*").as("hotCount"))
      .orderBy(desc("hotCount"))


    res.show()
  }

  def hotSubjectRecommendTop20(data:Dataset[AnswerWithRecommendations]): Unit = {
    //1.找到前20道热点题
    val hotTop20 = data.groupBy("question_id")
      .agg( count("*") as "hot" )
      .orderBy(desc("hot"))
      .limit(20)

    //2.根据前20道热点题，找到对应的推荐题
    val ridsDF = hotTop20.join(data.dropDuplicates("question_id"), "question_id")

    //3.对推荐列表进行切割，转换成Dataset
    val ridsDS: Dataset[Row] = ridsDF.select(explode(split(col("recommendations"), ",")) as "question_id")
      .dropDuplicates("question_id")

    //4.将ridsDS和data进行关联，得到每个推荐题目所属得科目
    val ridAndSid: DataFrame = ridsDS.join(data.dropDuplicates("question_id"), "question_id")

    //5.统计各个科目包含得推荐题目数据，并降序排序
    val res  = ridAndSid.groupBy("subject_id")
      .agg(count("*") as "rcount")
      .orderBy(desc("rcount"))

    res.show()
  }
}
